package cn.mjnxin.iot.gate.connector;

import java.util.concurrent.TimeUnit;

import cn.mjnxin.iot.base.domain.ProtocolConfig;
import cn.mjnxin.iot.base.log.LogWorker;
import cn.mjnxin.iot.base.util.CommonUtil;
import cn.mjnxin.iot.base.util.MixUtil;
import cn.mjnxin.iot.gate.coder.Gate2TerminalEncoder;
import cn.mjnxin.iot.gate.coder.Terminal2GateDecoder;
import cn.mjnxin.iot.gate.handle.GateAsClient2TerminalInHandler;
import cn.mjnxin.iot.transfer.ConnectionWacthDog;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * 网关作为客户端与Terminal相连
 * 
 * @author CLG 2023年10月10日
 * @version V1.0.0
 */
public class GateAsClient2Terminal implements AutoCloseable{
	private EventLoopGroup worker = new NioEventLoopGroup();
	private Integer pid;
	// 日志等级 0：无日志 1：报文长度日志 2：报文内容日志 3：所有日志
	private Integer logLevel;
	private String heartbeatPacket = null;

	private ConnectionWacthDog.Relogin loginTool = new ConnectionWacthDog.Relogin() {
		@Override
		public ByteBuf getLoginBuf(Integer gateNo, Integer pid, String LocalAddress) throws Exception {
			if (heartbeatPacket == null) {
				return null;
			}
			if (MixUtil.isHex(heartbeatPacket) && heartbeatPacket.length() % 2 == 1) {
				return null;
			}
        	ByteBuf out = CommonUtil.getByteBuf(heartbeatPacket.length() / 2);
        	out.writeBytes(MixUtil.hex2bytes(heartbeatPacket));
            return out;
		}

		@Override
		public void onConnectSuccess(String ip, Integer port, Integer pid) {
			LogWorker.logForce(String.format("网关连接Terminal %s:%s OK...", ip, port));
//		  CacheQueue.addHostChannel2LocalCache(ConstantValue.GATE_PID, ip + ConstantValue.CONNECTOR_LINE + masterPort, channelFuture.channel());
		}

		@Override
		public void onConnectFailed(String ip, Integer port, Integer pid) {
			LogWorker.logForce(String.format("网关连接等待Terminal %s:%s ...", ip, port));
		}
	};

	private final ConnectionWacthDog connectionWacthDog;
	private final Bootstrap bootstrap;
	private int heartbeat = 30;
	
	public GateAsClient2Terminal(Integer gateNo, String terminalAddress, int terminalPort, ProtocolConfig protocolConfig, boolean isOpenWatchDog) {
//		this.terminalAddress = terminalAddress;
//		this.terminalPort = terminalPort;
		this.heartbeatPacket = protocolConfig.getAutoHeartbeatPacket();
		this.pid = protocolConfig.getPid();
		this.logLevel = protocolConfig.getLogLevel();
        this.heartbeat = protocolConfig.getHeartbeat();
		connectionWacthDog = new ConnectionWacthDog(gateNo, pid, terminalAddress, terminalPort, isOpenWatchDog, loginTool) {
			@Override
			public ChannelHandler[] getChannelHandlers() {
				return new ChannelHandler[] { 
					new Terminal2GateDecoder(protocolConfig),
					new Gate2TerminalEncoder(protocolConfig.getLogLevel(), protocolConfig.getTerminalRepacker()), 
					this,
					new GateAsClient2TerminalInHandler(pid, logLevel, protocolConfig.getTerminalRepacker())
				};
			}

			@Override
			protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
				ctx.fireChannelRead(msg);
			}
		};
		bootstrap = connectionWacthDog.getBootstrap();
	}

	public void open() {
		bootstrap.group(worker)
			.channel(NioSocketChannel.class)
//			.option(ChannelOption.SO_KEEPALIVE, true)
//			.option(ChannelOption.TCP_NODELAY, true)
			/**
			 * 设置ByteBuf的高低水位线，原方法WRITE_BUFFER_HIGH_WATER_MARK，WRITE_BUFFER_LOW_WATER_MARK已经被废弃
			 * 由WRITE_BUFFER_WATER_MARK代替
			 * 加上WRITE_BUFFER_WATER_MARK配置后，往byteBuf写数据前需要判断isWritable
			 */
			.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(32 * 1024 * 1024, 64 * 1024 * 1024))
			.option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT)
			.handler(new ChannelInitializer<SocketChannel>() {
				@Override
				protected void initChannel(SocketChannel sc) throws Exception {
					sc.pipeline().addLast(new IdleStateHandler(0, 0, heartbeat, TimeUnit.SECONDS));
					sc.pipeline().addLast(connectionWacthDog.getChannelHandlers());
				}
			});
		
		connectionWacthDog.doConnect();
	}

	/**
	 * 关闭服务
	 */
	public void close() {
		CommonUtil.closeEventLoop(worker);
	}

	public Integer getpId() {
		return pid;
	}

	public void setpId(Integer pId) {
		this.pid = pId;
	}
}
